package com.b.service.impl;

import com.b.dao.LiveMapper;
import com.b.domain.Live;
import com.b.domain.User;
import com.b.repository.LiveRepository;
import com.b.repository.UserRepository;
import com.b.service.LiveService;
import com.b.vo.PageVo;
import com.b.vo.Push_Key;
import com.b.vo.Result;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchPage;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;

@Service
public class LiveServiceImpl implements LiveService{

    @Resource
    private LiveMapper liveMapper;

    @Resource
    private LiveRepository liveRepository;

    @Resource
    private UserRepository userRepository;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private ObjectMapper objectMapper;

//    @Resource
//    private RestTemplate restTemplate;

    private final String serviceAddress="rtmp://192.168.70.232:1935/";
    private final String stream_application_name = "live";
    private final String httpAddress = "http://192.168.70.232:8080/";
    private final String flv = ".flv";
    private final String hls = ".m3u8";

    private final String LIVE_PLAY_COUNT_KEY = "LivePlayCount";
    private final String LIVE_USER_VIEW_KEY = "LiveUserView";
    private final String Live_Looking_Count_Key = "LiveOnlineLooking";


    @Override
    public int deleteByPrimaryKey(Integer id) {
        return liveMapper.deleteByPrimaryKey(id);
    }

    @Override
    public int insert(Live record) {
        return liveMapper.insert(record);
    }

    @Override
    public int insertSelective(Live record) {
        return liveMapper.insertSelective(record);
    }

    @Override
    public Live selectByPrimaryKey(Integer id) {
        return liveMapper.selectByPrimaryKey(id);
    }

    @Override
    public int updateByPrimaryKeySelective(Live record) {
        return liveMapper.updateByPrimaryKeySelective(record);
    }

    @Override
    public int updateByPrimaryKey(Live record) {
        return liveMapper.updateByPrimaryKey(record);
    }

    @Override
    public Result createLive(Live live) {
        try {
            //检查当前用户是否已经创建直播
            Live userLive = liveRepository.findByUserIdAndLiveStatus(live.getUserId(), "直播中");
            if (userLive != null) {
                return new Result().setCode("500").setMsg("直播间创建失败!您已经开启直播间了!");
            }
            live.setPlayCount(0);
            live.setLookingCount(0);
            //创建直播间并返回推流地址和推流码
            String streamKey = UUID.randomUUID().toString() + live.getUserId();
            String pullAddress = serviceAddress + stream_application_name;
            live.setStreamKey(streamKey);
            live.setStreamApplicationName(stream_application_name);
            live.setPushAddress(serviceAddress+stream_application_name+"/"+streamKey);
            //flv拉流
            live.setPullAddress(httpAddress+stream_application_name+"/"+streamKey+flv);
            live.setLiveStartTime(new Date());
            live.setLiveStatus("直播中");
            System.out.println("live = " + live);
            liveMapper.insert(live);
            liveRepository.save(live);
            //创建返回的地址和key
            Push_Key pushKey = new Push_Key();
            pushKey.setStreamKey(streamKey);
            pushKey.setPushAddress(serviceAddress+stream_application_name+"/");
            return new Result().setCode("200").setMsg("直播间创建成功!").setData(pushKey);
        } catch (Exception e) {
            e.printStackTrace();
            return new Result().setCode("500").setMsg("服务器异常!直播间创建失败!");
        }
    }

    @Override
    public Result checkLiveStatus(Integer userId) {
        Live userLive = liveRepository.findByUserIdAndLiveStatus(userId, "直播中");
        if (userLive != null) {
            return new Result().setCode("500").setMsg("直播间创建失败!您已经开启直播间了!");
        }else {
            return new Result().setCode("200").setMsg("直播未开始");
        }
    }

    @Override
    public void importAll() {
        List<Live> all = liveMapper.getAll();
        liveRepository.saveAll(all);
    }

    @Override
    public Result endLive(Integer userId) {
        try {
            Live userLive = liveRepository.findByUserIdAndLiveStatus(userId, "直播中");
            if (userLive != null) {
                userLive.setLiveStatus("已结束");
                userLive.setLiveEndTime(new Date());
                //更新并清空缓存中的播放量数据
                HashOperations<String, Object, Object> opsForHash = stringRedisTemplate.opsForHash();
                String playerStr = (String) opsForHash.get(LIVE_PLAY_COUNT_KEY, userLive.getId().toString());
                Integer playCount = 0;
                if (playerStr != null) {
                    playCount = Integer.parseInt(playerStr);
                }
                userLive.setPlayCount(playCount);
                opsForHash.delete(LIVE_PLAY_COUNT_KEY, userLive.getId().toString());//删除缓存
                opsForHash.delete(Live_Looking_Count_Key,userLive.getId().toString());//删除观看人数缓存
                //更新数据库
                liveMapper.updateByPrimaryKeySelective(userLive);
                //更新es
                liveRepository.save(userLive);
                return new Result().setCode("200").setMsg("直播成功关闭!");
            }else {
                return new Result().setCode("500").setMsg("您暂无正在进行的直播!");
            }

        } catch (Exception e) {
            e.printStackTrace();
            return new Result().setCode("500").setMsg("服务器异常!直播未关闭");
        }
    }

    @Override
    public PageVo<Live> findLivesByTitleLike(Integer curPage, Integer pageSize, String keyword) {
        try {
            PageRequest pageRequest = PageRequest.of(curPage - 1, pageSize);
//            SearchPage<Live> liveSearchPage = liveRepository.findLivesByTitleLike(pageRequest, keyword);
            SearchPage<Live> liveSearchPage = liveRepository.findLiveByTitleLikeAndLiveStatus(pageRequest, keyword, "直播中");
            PageVo<Live> livePageVo = new PageVo<Live>(curPage, (int) liveSearchPage.getTotalElements(), pageSize);
            //定义pageList
            ArrayList<Live> liveArrayList = new ArrayList<>();
            List<SearchHit<Live>> content = liveSearchPage.getContent();
            for (SearchHit<Live> liveSearchHit : content) {
                System.out.println("liveSearchHit = " + liveSearchHit);
                Live live = liveSearchHit.getContent();
                //关联查询user
                User up = userRepository.findUserById(live.getUserId());
                //查询正在观看的人数
                HashOperations<String, Object, Object> opsForHash = stringRedisTemplate.opsForHash();
                Object o = opsForHash.get(Live_Looking_Count_Key, live.getId().toString());
                live.setAuthor(up);
                if (o != null) {//防止空指针
                    Integer LookingCount = Integer.valueOf((String) o) ;
                    System.out.println("LookingCount = " + LookingCount);
                    live.setLookingCount(LookingCount);
                    //封装成list数组
                    liveArrayList.add(live);
                }else {
                    Integer LookingCount = 0;
                    live.setLookingCount(LookingCount);
                    //封装成list数组
                    liveArrayList.add(live);
                }

            }
            livePageVo.setList(liveArrayList);
            return livePageVo;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    @Override
    public Result findLiveById(Integer liveId) {
        try {
            Optional<Live> live = liveRepository.findById(liveId);
            if (live.isPresent()) {
                Live liveHome = live.get();
                User author = userRepository.findUserById(liveHome.getUserId());
                liveHome.setAuthor(author);
                return new Result().setCode("200").setMsg("获取直播间详情成功!").setData(liveHome);
            }
            return new Result().setCode("500").setMsg("该直播间不存在!");
        } catch (Exception e) {
            e.printStackTrace();
            return new Result().setCode("500").setMsg("查询失败!服务器异常!");
        }
    }

    @Override
    public Result addPlyCount(Integer liveId,Integer userId) {
        // Redis哈希键（存储播放量）
        try {
            // 1. 检查直播间是否存在
            if (!liveRepository.existsById(liveId)) {
                return new Result().setCode("500").setMsg("当前直播间不存在!");
            }

            // 2. 防止同一用户短时间内重复计数（如1小时内）
            if (userId != null) {
                String userViewKey = liveId + ":" + userId;
                // 检查用户是否已计数，存在则直接返回
                if ("1".equals(stringRedisTemplate.opsForValue().get(userViewKey))) {
                    System.out.println("播放量已统计");
                    return new Result().setCode("200").setMsg("播放量已统计");
                }
                // 标记用户已计数，设置1小时过期
                stringRedisTemplate.opsForValue().set(userViewKey, "1", 1, TimeUnit.HOURS);
            }

            // 3. 核心：Redis原子递增播放量（并发安全）
            HashOperations<String, Object, Object> opsForHash = stringRedisTemplate.opsForHash();
            long newPlayCount = opsForHash.increment(LIVE_PLAY_COUNT_KEY, liveId.toString(), 1);

            System.out.println("播放量提升，当前值：" + newPlayCount);
            return new Result().setCode("200").setMsg("播放量提升").setData(newPlayCount);

        } catch (Exception e) {
            e.printStackTrace();
            return new Result().setCode("500").setMsg("服务器异常!");
        }
    }

    @Override
    public Result colLookingCount(Integer liveId,boolean isOnLooking) {
        HashOperations<String, Object, Object> opsForHash = stringRedisTemplate.opsForHash();
        Live live = liveRepository.findByIdAndLiveStatus(liveId, "直播中");
        if (live == null) {
            return new Result().setCode("500").setMsg("直播间不存在或未在直播中");
        }

        String hasKey = live.getId().toString();

        // 使用Redis原子操作更新计数（避免并发问题）
        long newCount;
        if (isOnLooking) {
            // 用户进入直播间，计数+1
            newCount = opsForHash.increment(Live_Looking_Count_Key, hasKey, 1);
        } else {
            // 用户离开直播间，计数-1（确保不会出现负数）
            newCount = opsForHash.increment(Live_Looking_Count_Key, hasKey, -1);
            if (newCount < 0) {
                // 防止负数，强制归零
                opsForHash.put(Live_Looking_Count_Key, hasKey, 0);
                newCount = 0;
            }
        }

        // 返回更新后的在线人数
        return new Result().setCode("200").setMsg("在线人数更新成功");
    }

    @Override
    public Result findMyLive(Integer userId) {
        Live myLiveHome = liveRepository.findByUserIdAndLiveStatus(userId, "直播中");
        if (myLiveHome == null) {
            return new Result().setCode("500").setMsg("查询失败!");
        }
        return new Result().setCode("200").setMsg("查询成功").setData(myLiveHome);
    }

    @Override
    public List<Live> getAll() {
        return liveMapper.getAll();
    }


    //定时方法:每1小时更新数据库的播放量数据
//    @Scheduled(cron = "0 0/1 * * * ?") //每分钟
    @Scheduled(cron = "0 0 0/1 * * ?") //一小时
//    @Scheduled(cron = "*/10 * * * * ?")
    public void updatePlyCount() {
        System.out.println("定时任务触发：每一小时执行一次,更新播放量");
        HashOperations<String, Object, Object> opsForHash = stringRedisTemplate.opsForHash();
        Map<Object, Object> map = opsForHash.entries(LIVE_PLAY_COUNT_KEY);
        for (Map.Entry<Object, Object> entry : map.entrySet()) {
            Integer liveId = Integer.valueOf((String) entry.getKey());
            Integer playCount = Integer.valueOf((String) entry.getValue());
            System.out.println("playCount = " + playCount);
            //更新相关数据
            Optional<Live> live = liveRepository.findById(liveId);
            if (live.isPresent()) {
                Live liveHome = live.get();
                liveHome.setPlayCount(playCount);
                //更新数据库和es
                liveMapper.updateByPrimaryKeySelective(liveHome);
                liveRepository.save(liveHome);
            }
        }
    }


}
